package com.leyantech.ecp.agenx.task;

import com.leyantech.ai.DialogueReply;
import com.leyantech.chaos.kafka.KafkaConsumerClient;
import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.ecp.agenx.common.ClusterVerticle;
import com.leyantech.ecp.agenx.task.di.FilterModule;
import com.leyantech.ecp.agenx.task.di.TaskModule;

import com.google.common.collect.Lists;
import com.google.inject.Inject;
import com.google.inject.Module;
import io.vertx.core.json.JsonObject;
import io.vertx.core.spi.cluster.ClusterManager;

import java.util.List;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-10-25.
 */
public class TaskVerticle extends ClusterVerticle {

  @Inject
  private KafkaConsumerClient<String, JsonObject> taskConsumer;

  @Inject
  private KafkaConsumerClient<String, DialogueReply> dialogueReplyConsumer;

  @Inject
  KafkaProducerStringClient producer;

  private void startHealthCheckServer() {
    vertx.createHttpServer()
        .requestHandler(request -> request.response().send("ok"))
        .listen(8080);
  }

  @Override
  public void start() {
    taskConsumer.start();
    dialogueReplyConsumer.start();
    addRunnableCloseHook(taskConsumer::destroy);
    addRunnableCloseHook(dialogueReplyConsumer::destroy);
    addRunnableCloseHook(producer::destroy);
    startHealthCheckServer();
  }

  @Override
  protected ClusterManager configClusterManager() {
    return null;
  }

  @Override
  protected List<Module> dependentModules() {
    return Lists.newArrayList(new FilterModule(), new TaskModule());
  }

  public static void main(String[] args) {
    new TaskVerticle().startVerticle();
  }

}
